package org.activequant.data.util;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;

import org.activequant.core.domainmodel.marketscanner.MarketScanSeriesSpecification;
import org.activequant.core.domainmodel.marketscanner.ScanCriteria;
import org.activequant.core.domainmodel.marketscanner.ScanDataEntity;
import org.activequant.core.types.TimeFrame;
import org.activequant.core.types.TimeStamp;
import org.activequant.data.retrieval.IScanSeriesDataIteratorSource;
import org.activequant.data.retrieval.ScanSubscriptionSourceBase;
import org.activequant.util.pattern.events.IEventSink;
import org.apache.log4j.Logger;

/**
* An abstract class. ScanReplayServiceBase&lt;T extends ScanDataEntity&lt;T&gt;&gt; extends ScanSubscriptionSourceBase&lt;T&gt; implements IEventSink&lt;T&gt;. 
* Defines the DispatchingSubscription class. Holds the following associated variables:
* <ul>
* <li>events(Map&lt;MarketScanSeriesSpecification,IEventSink&lt;T&gt;&gt;)</li>
* <li>currentTimeStamp(AtomicReference&lt;TimeStamp&gt;)</li>
* <li>startTimeStamp(TimeStamp)</li>
* <li>endTimeStamp(TimeStamp)</li>
* <li>seriesSource(IScanSeriesDataIteratorSource&lt;T&gt;)</li>
* <li>activeSeries(Set&lt;MarketScanSeriesSpecification&gt;)</li>
* </ul>
* It also holds the following inherited variables:
* <ul>
* <li>activeBackends(Map&lt;Topic, Subscription&gt;)</li>
* <li>activeSubscriptions(Queue&lt;SubscriptionDelegate&gt;)</li>
* </ul>
*  @author Dimitar Misev
*/
public abstract class ScanReplayServiceBase<T extends ScanDataEntity<T>> extends ScanSubscriptionSourceBase<T> implements IEventSink<T> {

	private final Logger log = Logger.getLogger(getClass()); 
	/**
	 * An empty ScanReplayServiceBase&lt;T extends ScanDataEntity&lt;T&gt;&gt; (extends ScanSubscriptionSourceBase&lt;T&gt; implements IEventSink&lt;T&gt;) constructor
	 */
	public ScanReplayServiceBase() {
		super();
	}
	/**
	 * an abstract method
	 * @param entity
	 * @return
	 */
	protected abstract MarketScanSeriesSpecification inferSeriesSpecification(T entity); 
	/**
	 * private final Map&lt;MarketScanSeriesSpecification,IEventSink&lt;T&gt;&gt; events = new ConcurrentHashMap&lt;MarketScanSeriesSpecification,IEventSink&lt;T&gt;&gt;();
	 */
	private final Map<MarketScanSeriesSpecification,IEventSink<T>> events = new ConcurrentHashMap<MarketScanSeriesSpecification,IEventSink<T>>();
	/**
	 * <strong>1.</strong> Maps the given specs(MarketScanSeriesSpecification)-sink(IEventSink&lt;T&gt;) pair into the associated events(Map&lt;MarketScanSeriesSpecification,IEventSink&lt;T&gt;&gt;)<br/>
	 * <strong>2.</strong> Creates a copy specs(MarketScanSeriesSpecification) of the given specs(MarketScanSeriesSpecification), except it sets its start(TimeStamp) and end(TimeStamp) with the value of the
	 * associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) and endTimeStamp(TimeStamp)<br/>
	 * <strong>3.</strong> Adds a copy of the given specs(MarketScanSeriesSpecification) into the associated activeSeries(Set&lt;MarketScanSeriesSpecification&gt;). If that MarketScanSeriesSpecification element already 
	 * existed there, it returns without doing anything else. Otherwise it gets an iterable(Iterable&lt;T&gt;) for the copy specs(MarketScanSeriesSpecification) using the 
	 * associated seriesSource(IScanSeriesDataIteratorSource&lt;T&gt;). Then it invokes the abstract <code>processFetchedIterator(iterable.iterator())</code>
	 * @param specs
	 * @param sink
	 */
	private void addEventSink(MarketScanSeriesSpecification specs, IEventSink<T> sink) {
		events.put(specs, sink);
		fetchSeries(specs);
	}
	/**
	 * removes the mapping for the given specs(MarketScanSeriesSpecification) from the associated events(Map&lt;MarketScanSeriesSpecification,IEventSink&lt;T&gt;&gt;)
	 * @param specs
	 */
	private void removeEventSink(MarketScanSeriesSpecification specs) {
		events.remove(specs);
	}
	/**
	 * A nested class. Extends Subscription implements IEventSink&lt;T&gt;. Holds the following associated variable:
	 * <ul>
	 * <li>specs(MarketScanSeriesSpecification)</li>
	 * </ul>
	 * It also holds the following inherited variables:
	 * <ul>
	 * <li>isActive(AtomicBoolean)</li>
	 * <li>refCount(AtomicInteger)</li>
	 * <li>scanCriteria(ScanCriteria)</li>
	 * <li>event(Event&lt;T&gt;)</li>
	 * </ul>
	 * @author Dimitar
	 *
	 */
	class DispatchingSubscription extends Subscription implements IEventSink<T> {
		/**
		 * private final MarketScanSeriesSpecification specs;
		 */
		private final MarketScanSeriesSpecification specs;
		/**
		 * Constructs a DispatchingSubscription (extends Subscription implements IEventSink&lt;T&gt;) using the given ss(MarketScanSeriesSpecification) to set its
		 * associated specs(MarketScanSeriesSpecification)
		 * @param ss
		 */
		public DispatchingSubscription(MarketScanSeriesSpecification ss) {
			specs = ss;
		}
		/**
		 * <strong>1.</strong> Maps this DispatchingSubscription instance into the associated events(Map&lt;MarketScanSeriesSpecification,IEventSink&lt;T&gt;&gt;) using the associated specs(MarketScanSeriesSpecification) as key<br/>
		 * <strong>2.</strong> Creates a copy specs(MarketScanSeriesSpecification) of the associated specs(MarketScanSeriesSpecification), except it sets its start(TimeStamp) and end(TimeStamp) with the value of the
		 * associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) and endTimeStamp(TimeStamp)<br/>
		 * <strong>3.</strong> Adds a copy of the associated specs(MarketScanSeriesSpecification) into the associated activeSeries(Set&lt;MarketScanSeriesSpecification&gt;). If that MarketScanSeriesSpecification element already 
		 * existed there, it returns without doing anything else. Otherwise it gets an iterable(Iterable&lt;T&gt;) for the copy specs(MarketScanSeriesSpecification) using the 
		 * associated seriesSource(IScanSeriesDataIteratorSource&lt;T&gt;). Then it invokes the abstract <code>processFetchedIterator(iterable.iterator())</code>
		 */
		@Override
		protected void handleActivate() { 
			addEventSink(specs, this);
		}
		/**
		 * removes the mapping for the associated specs(MarketScanSeriesSpecification) from the events(Map&lt;MarketScanSeriesSpecification,IEventSink&lt;T&gt;&gt;)
		 */
		@Override
		protected void handleCancel() { 
			removeEventSink(specs);
		}
		/**
		 * invokes <code>event.fire(data)</code> on the inherited event(Event&lt;T&gt;) using the given e(T)
		 */
		public void fire(T e) {
			super.fireEvent(e);
		}
	}
	/**
	 * <strong>1.</strong> Creates shortSs(MarketScanSeriesSpecification) using the given spec(ScanCriteria) and timeFrame(TimeFrame) to set its associated scanCriteria(ScanCriteria) and timeFrame(TimeFrame)<br/>
	 * <strong>2.</strong> Creates and returns a DispatchingSubscription(extends Subscription implements IEventSink&lt;T&gt;) using that shortSs(MarketScanSeriesSpecification) to set its associated specs(MarketScanSeriesSpecification)
	 */
	@Override
	protected DispatchingSubscription createSubscription(ScanCriteria spec, TimeFrame timeFrame) {
		
		MarketScanSeriesSpecification shortSs = new MarketScanSeriesSpecification(spec, timeFrame);
		
		return new DispatchingSubscription(shortSs);
	}
	/**
	 * <strong>1.</strong> Finds the spec(MarketScanSeriesSpecification) for the given e(T) by invoking the abstract <code>inferSeriesSpecification(e)</code> method.<br/>
	 * <strong>2.</strong> Gets the sink(IEventSink&lt;T&gt;) mapped under that spec(MarketScanSeriesSpecification) into the associated events(Map&lt;MarketScanSeriesSpecification,IEventSink&lt;T&gt;&gt;)<br/>
	 * <strong>3.</strong> Fires the given e(T) on that sink(IEventSink&lt;T&gt;) by invoking the <code>sink.fire(e)</code> method.
	 */
	public void fire(T e) {
		MarketScanSeriesSpecification spec = inferSeriesSpecification(e);
		IEventSink<T> sink = events.get(spec);
		if(sink != null) {
			try {
				sink.fire(e);
			} catch(Exception ex) {
				ex.printStackTrace();
				log.error(ex);
			}
		}
	}
	
	/**
	 * Processes the iterator over the fetched market data entities.
	 * Subclass must implement it and do something with the data. Typically, subclass will add this to the
	 * mergeSort to merge and align with the other data streams.<br/>
	 * An abstract method
	 * @param iterator
	 */
	protected abstract void processFetchedIterator(Iterator<T> iterator);
	/**
	 * private final AtomicReference&lt;TimeStamp&gt; currentTimeStamp = new AtomicReference&lt;TimeStamp&gt;();
	 */
	private final AtomicReference<TimeStamp> currentTimeStamp = new AtomicReference<TimeStamp>();
	/**
	 * private TimeStamp startTimeStamp;
	 */
	private TimeStamp startTimeStamp;
	
	/**
	 * Start of the historical interval.<br/>
	 * returns the associated startTimeStamp(TimeStamp)
	 * @return start.
	 */
	public TimeStamp getStartTimeStamp() {
		return startTimeStamp;
	}
	/**
	 * <strong>1.</strong> Sets the associated startTimeStamp(TimeStamp) with the given val(TimeStamp)<br/>
	 * <strong>2.</strong> Sets the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) with the given val(TimeStamp)
	 * @param val
	 */
	public void setStartTimeStamp(TimeStamp val) {
		startTimeStamp = val;
		currentTimeStamp.set(startTimeStamp);
	}
	/**
	 * private TimeStamp endTimeStamp;
	 */
	private TimeStamp endTimeStamp;
	
	/**
	 * End of the historical interval.<br/>
	 * returns the associated endTimeStamp(TimeStamp)
	 * @return end.
	 */
	public TimeStamp getEndTimeStamp() {
		return endTimeStamp;
	}
	/**
	 * sets the associated endTimeStamp(TimeStamp) with the given val(TimeStamp)
	 * @param val
	 */
	public void setEndTimeStamp(TimeStamp val) {
		endTimeStamp = val;
	}
	/**
	 * private IScanSeriesDataIteratorSource&lt;T&gt; seriesSource;
	 */
	private IScanSeriesDataIteratorSource<T> seriesSource;
	
	/**
	 * Historic data source for quotes.<br/>
	 * returns the associated seriesSource(IScanSeriesDataIteratorSource&lt;T&gt;)
	 * @return data source.
	 */
	public IScanSeriesDataIteratorSource<T> getSeriesDataSource() {
		return seriesSource;
	}
	/**
	 * Sets historic data source for quotes.<br/>
	 * sets the associated seriesSource(IScanSeriesDataIteratorSource&lt;T&gt;) with the given val(IScanSeriesDataIteratorSource&lt;T&gt;)
	 * @param val data source.
	 */
	public void setSeriesDataSource(IScanSeriesDataIteratorSource<T> val) {
		seriesSource = val;
	}
	/**
	 * private final Set&lt;MarketScanSeriesSpecification&gt; activeSeries = new HashSet&lt;MarketScanSeriesSpecification&gt;();
	 */
	private final Set<MarketScanSeriesSpecification> activeSeries = new HashSet<MarketScanSeriesSpecification>();
	/**
	 * <strong>1.</strong> Creates a copy specs(MarketScanSeriesSpecification) of the given ss(MarketScanSeriesSpecification), except it sets its start(TimeStamp) and end(TimeStamp) with the value of the
	 * associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) and endTimeStamp(TimeStamp)<br/>
	 * <strong>2.</strong> Adds a copy of the given ss(MarketScanSeriesSpecification) into the associated activeSeries(Set&lt;MarketScanSeriesSpecification&gt;). If that MarketScanSeriesSpecification element already 
	 * existed there, it returns without doing anything else. Otherwise it gets an iterable(Iterable&lt;T&gt;) for the copy specs(MarketScanSeriesSpecification) using the 
	 * associated seriesSource(IScanSeriesDataIteratorSource&lt;T&gt;). Then it invokes the abstract <code>processFetchedIterator(iterable.iterator())</code>
	 * @param ss
	 */
	private void fetchSeries(MarketScanSeriesSpecification ss) {
		if(startTimeStamp == null || endTimeStamp == null) {
			throw new IllegalStateException("backtester must have start and end dates set");
		}
		try {
			MarketScanSeriesSpecification specs = new MarketScanSeriesSpecification(ss.getScanCriteria(), ss.getTimeFrame());
			if(activeSeries.add(new MarketScanSeriesSpecification(specs)) == false) {
				log.info("requested stream is already active: " + specs);
				return;
			}
			specs.setStartTimeStamp(currentTimeStamp.get());
			specs.setEndTimeStamp(endTimeStamp);
			//log.info("fetching data for: " + specs + " from " + seriesSource.getClass());
			Iterable<T> iterable = seriesSource.fetch(specs);
			processFetchedIterator(iterable.iterator());
		} catch(Exception ex) {
			ex.printStackTrace();
			log.error(ex);
		}
	}
}
